import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import java.util.Collections;
import java.util.TreeSet;
import java.util.SortedSet;
import java.io.IOException;

public class confirm_async{
	public static void main(String[] args) throws Exception {
            ConnectionFactory cf = new ConnectionFactory();
            cf.setHost("192.168.1.105");
            cf.setUsername("liwenbin");
            cf.setPassword("123456");
            Connection connection = cf.newConnection();
            Channel channel = connection.createChannel();
            SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
            channel.confirmSelect();
            channel.addConfirmListener(new ConfirmListener() {
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Ack,SeqNo:" + deliveryTag + "multiple:" + multiple);
                    if (multiple) {
                        confirmSet.headSet(deliveryTag + 1).clear();
                    } else {
                        confirmSet.remove(deliveryTag);
                    }
                }
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
                    if (multiple) {
                        confirmSet.headSet(deliveryTag + 1).clear();
                    } else {
                        confirmSet.remove(deliveryTag);
                    }
                }
            });
            int i = 0;
            while (i++ < 100) {
                long nextSeqNo = channel.getNextPublishSeqNo();
                String message = "message" + i;
                String QUEUE_NAME = "hello";
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                confirmSet.add(nextSeqNo);
             }
//             channel.close();
//             connection.close();
	}
}

